package com.example.channel

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.actor
import kotlinx.coroutines.channels.produce
import org.junit.Test


/**
 * 通道的创建
 * 1)通过Channel创建对象
 *
 * @author zhouronghua
 * @time 2022/1/9 7:19 下午
 */
class ChannelCreateTest {

    /**
     * 创建通道
     * 创建通道，然后通过通道发送元素给其他协程
     *
     * @author zhouronghua
     * @time 2022/1/9 8:36 下午
     */
    @Test
    fun test_channel_create() = runBlocking {
        println("测试通道创建和发射元素")
        val channel = Channel<Int>()
        // 生产者
        val producer = GlobalScope.launch {
            var i = 0
            while (true) {
                delay(100)
                channel.send(++i)
                println("producer send channel $i")

            }
        }
        // 消费者
        val consumer = GlobalScope.launch {
            while (true) {
                delay(150)
                val element = channel.receive()
                println("consumer receive channel $element")
            }
        }
        joinAll(producer, consumer)
    }

    /**
     * 通道遍历
     * 通过迭代器遍历元素
     *
     * @author zhouronghua
     * @time 2022/1/9 8:36 下午
     */
    @Test
    fun test_channel_iterator() = runBlocking {
        println("测试通道元素迭代")
        val channel = Channel<Int>()
        // 生产者
        val producer = GlobalScope.launch {
            for (i in 1..100) {
//                delay(100)
                channel.send(i)
                println("producer send channel $i")
            }
            // 关闭channel
            channel.close()
        }
        // 消费者
        val consumer = GlobalScope.launch {
//            var it = channel.iterator()
//            while (it.hasNext()) {
//                println("consumer receive channel ${it.next()}")
//            }
            for (it in channel) {
                delay(300)
                println("consumer receive channel $it")
            }
        }
        joinAll(producer, consumer)
    }

    /**
     * 测试通过produce创建生产者和发送消息
     *
     * @author zhouronghua
     * @time 2022/1/9 8:58 下午
     */
    @Test
    fun test_channel_produce() = runBlocking {
        val receiveChannel = produce<Int> {
            repeat(100) {
                delay(100)
                send(it)
                println("produce send $it")
            }
        }
        val consumer = GlobalScope.launch {
            for (it in receiveChannel) {
                delay(300)
                println("consumer receive channel $it")
            }
        }
        consumer.join()
    }

    /**
     * 测试通过actor创建消费者
     *
     * @author zhouronghua
     * @time 2022/1/9 8:58 下午
     */
    @Test
    fun test_channel_actor() = runBlocking {
        val sendChannel = actor<Int> {
            // 创建消费者接受消息
            while (true) {
                val element = receive()
                println("consumer receive channel $element")
            }
        }
        // 创建生产者通过消费者的通道发送消息
        val producer = GlobalScope.launch {
            for (i in 1..100) {
                // 每个100秒通过消费者的通道发送消息
                delay(100)
                sendChannel.send(i)
                println("sendChannel 发送 $i")
            }
            sendChannel.close()
        }
        producer.join()
    }

    /**
     * 通道通过close关闭
     * 说明：
     * 如果channel调用close关闭通道，他会立即停止接收新元素，也就是isClosedForSend会立即返回true
     * 如果缓冲区中还有元素没有处理完，因此需要所有元素读取完isClosedForReceive才会返回true
     * Channel的关闭最好是主导方来实现关闭
     *
     * @author zhouronghua
     * @time 2022/1/9 9:18 下午
     */
    @Test
    fun test_channel_close() = runBlocking {
        println("测试通道关闭")
        val channel = Channel<Int>()
        // 生产者
        val producer = GlobalScope.launch {
            var i = 0
            List(3) {
                channel.send(++i)
                println("producer send channel $i")
            }
            // 关闭通道
            println("关闭通道")
            channel.close()
            println("通道 isClosedForSend ${channel.isClosedForSend} isClosedForReceive ${channel.isClosedForReceive} ")
        }
        // 消费者
        val consumer = GlobalScope.launch {
            try {
                while (!channel.isClosedForReceive) {
                    delay(150)
                    val element = channel.receive()
                    println("consumer receive channel $element")
                }
            } catch (e: Throwable) {
                println("Caught: $e")
            } finally {
                println("consumer通道 isClosedForSend ${channel.isClosedForSend} isClosedForReceive ${channel.isClosedForReceive} ")
            }
        }
        joinAll(producer, consumer)
    }
}